extern crate core;

mod context;
mod event_entity;
mod event_handle;
mod interface;
mod schedule;

pub use async_trait::async_trait;
pub use context::Context;
pub use event_entity::*;
pub use event_handle::Handle;
pub use interface::*;
pub use schedule::EventSchedule;

#[cfg(test)]
mod test {
    use crate::{Context, Event, EventSchedule};
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Arc;

    struct Message<V> {
        id: u64,
        value: V,
    }
    impl<V> Message<V> {
        fn new(id: u64, value: V) -> Message<V> {
            Self { id, value }
        }
    }
    fn generate_event_entity() -> Event<EventSchedule> {
        EventSchedule::default()
            .register(|ctx: Context, mut msg: Message<String>| {
                // println!("this is first Message<String> handler ID[{}]--> {}",msg.id,msg.value);
                msg.value.push_str("->one");
                ctx.next(msg) // no need async
            })
            .register(|ctx: Context, mut msg: Message<String>| async {
                // println!("this is second Message<String> handler ID[{}]--> {}",msg.id,msg.value);
                msg.value.push_str("->two");
                ctx.next(msg).await //all is async
            })
            .register(|ctx: Context, mut msg: Message<i64>| {
                // println!("this is third Message<i64> handle  ID[{}]--> {}",msg.id,msg.value);
                msg.value = 3;
                async move {
                    ctx.next(msg).await //part is async
                }
            })
            .build()
    }

    #[tokio::test]
    async fn test_message() {
        let event = generate_event_entity();
        let res = event
            .launch(
                Context::default(),
                Message::new(1, "hello world".to_string()),
            )
            .await
            .expect("first test handle failed");
        assert_eq!(
            res.value.as_str(),
            "hello world->one->two",
            "Message<String> result assert failed"
        );
        let res = event
            .launch(Context::default(), Message::new(1, 1i64))
            .await
            .expect("second test failed");
        assert_eq!(res.value, 3i64, "Message<i64> result assert failed")
    }
    #[tokio::test]
    async fn test_pressure_single_thread() {
        let event = generate_event_entity();

        let stopwatch = std::time::Instant::now();
        let max = 1_00_0000;
        for i in 0..max {
            let res = event
                .launch(
                    Context::default(),
                    Message::new(1, "hello world".to_string()),
                )
                .await
                .expect("first test handle failed");
            assert_eq!(
                res.value.as_str(),
                "hello world->one->two",
                "Message<String> result assert failed"
            );
            if i % 10_0000 == 0 {
                println!("[{}]->running", i / 100000);
            }
        }
        let t = stopwatch.elapsed();
        println!("run {} times use: {}s", max, t.as_secs());
    }
    #[tokio::test(flavor = "multi_thread", worker_threads = 12)]
    async fn test_concurrency_thread_pool() {
        let event = Arc::new(generate_event_entity());
        let stopwatch = std::time::Instant::now();
        let count = Arc::new(AtomicUsize::new(1000 * 1000));
        for _ in 0..1000 {
            let event = event.clone();
            let count = count.clone();
            tokio::spawn(async move {
                for _ in 0..1000 {
                    let res = event
                        .launch(
                            Context::default(),
                            Message::new(1, "hello world".to_string()),
                        )
                        .await
                        .expect("first test handle failed");
                    assert_eq!(
                        res.value.as_str(),
                        "hello world->one->two",
                        "Message<String> result assert failed"
                    );
                    count.fetch_sub(1, Ordering::Relaxed);
                }
            });
        }
        while count.load(Ordering::Relaxed) > 0 {
            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        let t = stopwatch.elapsed();
        println!(
            "concurrency 1000 execute 1000 times use: {}ms",
            t.as_millis()
        );
    }
}
